package com.sfang.common.rocketmq;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.sfang.common.rocketmq.config.MqConfig;
import org.apache.rocketmq.client.exception.MQClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.core.env.Environment;

import org.springframework.stereotype.Component;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.function.Consumer;

@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class RocketmqUtil implements ApplicationRunner {
    @Autowired
    Environment env;
    @Autowired
    private MqConfig mqConfig;
    @Autowired
    ProducerBean rocketmqProducerBean;
    @Value("${jt.current.profile:}")
    private String profile;

    static String enviromentType;
    static Environment environment;
    static ProducerBean rocketmqProducer;
    static MqConfig rocketmqConfig;
    private static Logger LOGGER = LoggerFactory.getLogger(RocketmqUtil.class);


    @PostConstruct
    public void init() {
        this.environment = env;

        if (environment != null && environment.getProperty("env") != null)
            this.enviromentType = environment.getProperty("env");
        else if (environment != null && environment.getProperty("spring.cloud.config.profile") != null)
            this.enviromentType = environment.getProperty("spring.cloud.config.profile");
        else if (profile != null && profile.length() > 0)
            this.enviromentType = profile;

    }

    public static <T> Boolean send(String topic, String tags, T value) {
        try {
            if (enviromentType != null && !enviromentType.equals("prod")) {
                topic += enviromentType.toUpperCase();
            }

            String rocketmqKey = MDC.get("traceId") != null
                    ? MDC.get("traceId")
                    : UUID.randomUUID().toString().replace("-", "");

            if (MDC.get("traceId") == null)
                MDC.put("traceId", rocketmqKey);

            Message message = new Message(topic, tags, rocketmqKey, JSON.toJSONString(value).getBytes());

            SendResult sendResult = rocketmqProducer.send(message);

            String messageValue = "";

            try {
                messageValue = JSON.toJSONString(message);
            } catch (Exception e) {
                messageValue = String.valueOf(message);
            }

            if (sendResult != null) {
                LOGGER.info(String.format("rocketmq send message success,topic:%s tags:%s keyId:%s messageId:%s traceId:%s message:%s",
                        sendResult.getTopic(), tags, rocketmqKey, sendResult.getMessageId(), MDC.get("traceId"), messageValue));
                return true;
            } else {
                LOGGER.error(String.format("rocketmq send message failed,topic:%s tags:%s keyId:%s traceId:%s message:%s",
                        topic, tags, rocketmqKey, MDC.get("traceId"), messageValue));
                return false;
            }
        } catch (Exception e) {
            LOGGER.error(String.format("rocketmq send message error,topic:%s tags:%s message:%s traceId:%s error:%s",
                    topic,
                    tags,
                    JSON.toJSONString(value),
                    MDC.get("traceId"),
                    JSON.toJSONString(e)
            ));
            return false;
        }
    }

    public static <T> void consume(String topic, String topicConsumerGroup, Consumer<T> consumer, Class<T> clazz) throws MQClientException {
        if (enviromentType != null && !enviromentType.equals("prod")) {
            topic += enviromentType.toUpperCase();
            topicConsumerGroup += enviromentType.toUpperCase();
        }
        consumer(topic, topicConsumerGroup, consumer, null, clazz);
    }

    public static <T> void consumeWithTags(String topic, String topicConsumerGroup, Consumer<Tuple2<T, String>> consumer, Class<T> clazz) throws MQClientException {
        if (enviromentType != null && !enviromentType.equals("prod")) {
            topic += enviromentType.toUpperCase();
            topicConsumerGroup += enviromentType.toUpperCase();
        }
        consumer(topic, topicConsumerGroup, null, consumer, clazz);
    }

    private static <T> void consumer(String topic, String topicConsumerGroup, Consumer<T> consumer, Consumer<Tuple2<T, String>> consumerWithTags, Class<T> clazz) throws MQClientException {

        ConsumerBean consumerBean = new ConsumerBean();
        //配置文件
        Properties properties = rocketmqConfig.getMqPropertie();
        properties.setProperty(PropertyKeyConst.GROUP_ID, topicConsumerGroup);
        //将消费者线程数固定为20个 20为默认值
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        consumerBean.setProperties(properties);
        //订阅关系
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
        consumerBean.setSubscriptionTable(subscriptionTable);
        Subscription subscription = new Subscription();
        subscription.setTopic(topic);
        subscription.setExpression(rocketmqConfig.getTag());
        subscriptionTable.put(subscription, new MessageListener() {

            @Override
            public Action consume(Message messageExt, ConsumeContext consumeContext) {
                String traceId = messageExt.getKey();
                if (traceId != null)
                    MDC.put("traceId", traceId);

                String bodyValue = new String(messageExt.getBody());
                LOGGER.info(String.format("rocketmq consume msg:%s group:%s msgId:%s message:%s tag:%s traceId:%s",
                        topic,
                        topicConsumerGroup,
                        messageExt.getMsgID(),
                        bodyValue,
                        messageExt.getTag(),
                        traceId));

                try {
                    T t = clazz.getSimpleName().equalsIgnoreCase("String")
                            ? (T) bodyValue
                            : JSON.parseObject(bodyValue, clazz);

                    if (consumer != null)
                        consumer.accept(t);

                    if (consumerWithTags != null)
                        consumerWithTags.accept(Tuples.of(t, messageExt.getTag()));

                } catch (Exception e) {
                    LOGGER.error(String.format("rocketmq consume error:%s msgId:%s tags:%s traceId:%s", topic, messageExt.getMsgID(), messageExt.getTag(), traceId), e);
                    return Action.ReconsumeLater;
                }

                LOGGER.info(String.format("rocketmq consume ack success:%s msgId:%s message:%s tags:%s traceId:%s",
                        topic,
                        messageExt.getMsgID(),
                        bodyValue,
                        messageExt.getTag(),
                        traceId));
                return Action.CommitMessage;
            }
        });
        consumerBean.start();
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        rocketmqProducer = rocketmqProducerBean;
        rocketmqConfig = mqConfig;

    }
}
